package job

import (
	"context"
	"git.multiverse.io/eventkit/dxc-server/client"
	"git.multiverse.io/eventkit/dxc-server/config"
	"git.multiverse.io/eventkit/dxc-server/mgmt/abnormal_txn"
	"git.multiverse.io/eventkit/dxc-server/models"
	"git.multiverse.io/eventkit/dxc-server/repo"
	"git.multiverse.io/eventkit/dxc-server/service"
	"git.multiverse.io/eventkit/kit/constant"
	"git.multiverse.io/framework/log"
	jsoniter "github.com/json-iterator/go"
	"time"
)

const (
	TransactionRecoverWorkerNumber = 32
)

var json = jsoniter.ConfigCompatibleWithStandardLibrary

// @Desc job txn
func StartTransactionRecoverJob() {
	log.Infos("Start transaction recover job...")
	defer log.Infos("Started transaction recover job successfully!")

	for i := 0; i < TransactionRecoverWorkerNumber; i++ {
		go transactionRecoverWorker(i)
	}
}

func deleteTransactionByXid(repository repo.Repository, workerId int, rootXid string) {
	log.Debugsf("Recovery worker[%d]:deleteTransaction, root xid:[%s]!", workerId, rootXid)
	if _, e := repository.DeleteTxn(rootXid); e != nil {
		log.Errorsf("Recovery worker[%d]:Remove root transaction{RootXid:[%s]} failed, error:[%v]", workerId, rootXid, e)
	}
}

func recoverConfirmingTransactions(repository repo.Repository, callbackService service.CallbackService, workerId int, root *models.RootTransaction, pair models.Pair) {
	log.Debugsf("recover Confirming Transactions, root:[%++v]", root)
	if !root.IsTheFirstTimeToRecover() && root.IsRecoverTimeExceededLimit(config.Cfg.Job.DurationSeconds) {
		log.Errorsf("Recovery worker[%d]: The Total time cost of the current transaction[%++v] for confirm has exceeded the limit[%d](seconds),try to report to DTS management",
			workerId, root, config.Cfg.Job.DurationSeconds)
		//if successfully synchronized transactions to DTS Management, then delete transaction in DTS agent
		reason := "recover confirm transaction fail, The total time cost of the current transaction for confirm has exceeded the limit"
		if err := abnormal_txn.ReportTransactionToDTSManagement(root, reason); nil != err {
			log.Errorsf("recoverConfirmingTransactions,report transaction to dxc management failed, error=%++v", err)
		} else {
			deleteTransactionByXid(repository, workerId, root.RootXid)
		}
		return
	}
	success := true
	branches := make([]*models.Transaction, 0)
	callbackFails := make([]*models.Transaction, 0)
	for _, branch := range root.Branches {
		if branch.BranchTxnStat == "1" {
			continue
		}
		if !root.IsTheFirstTimeToRecover() && branch.IsRecoverTimeExceededLimit(config.Cfg.Job.MinIntervalSeconds) {
			success = false
			callbackFails = append(callbackFails, branch)
			continue
		}
		if branch.IsRoot() && branch.IsClientSDKVersion(constant.ClientSDKVersionV1) {
			// The current version of client is V1, If the branch is root then confirm/cancel will execute by client SDK
			log.Debugsf("Skip confirm root transaction[%s]!", branch.RootXid)
			continue
		}
		branches = append(branches, branch)
		if err := callbackService.BranchTxnCallback(pair.SpanCtx, branch, true); err != nil {
			log.Errorsf("[Recover] Branch transaction confirm failed, err:%s", err)
			success = false
			branch.RecoveryTimes = branch.RecoveryTimes + 1
			branch.BranchTxnLastUpdateTime = time.Now()
			callbackFails = append(callbackFails, branch)
		}
	}
	if !success {
		if len(callbackFails) > 0 {
			firstTimeCallback := true
			if pair.Type != models.CallbackFailedRecordsRecover {
				firstTimeCallback = false
			}
			_, err := repository.CallbackFail(root.RootXid, branches, firstTimeCallback)
			if err != nil {
				log.Errorsf("[Recover] confirm callback failed, error=%++v", err)
			}
			time.Sleep(1 * time.Second)
			pair.SpanCtx.TimeoutMilliseconds = 30000
			pair.Type = models.CallbackFailedRecordsRecover
			models.Queue.Push(pair)
		}
		return
	}
	deleteTransactionByXid(repository, workerId, root.RootXid)
}

func recoverCancellingTransactions(repository repo.Repository, callbackService service.CallbackService, workerId int, root *models.RootTransaction, pair models.Pair) {
	log.Debugsf("recover Cancelling Transactions, root:[%++v]", root)
	if !root.IsTheFirstTimeToRecover() && root.IsRecoverTimeExceededLimit(config.Cfg.Job.DurationSeconds) {
		log.Errorsf("Recovery worker[%d]: The Total time cost of the current transaction[%++v] for cancel has exceeded the limit[%d](seconds),try to report to DTS management",
			workerId, root, config.Cfg.Job.DurationSeconds)
		// if successfully synchronized transactions to DTS Management, then delete transaction in DTS agent
		reason := "recover cancel transaction fail, The total time cost of the current transaction for cancel has exceeded the limit"
		if err := abnormal_txn.ReportTransactionToDTSManagement(root, reason); nil != err {
			log.Errorsf("recoverConfirmingTransactions,report transaction to dxc management failed, error=%++v", err)
		} else {
			deleteTransactionByXid(repository, workerId, root.RootXid)
		}
		return
	}
	success := true
	branches := make([]*models.Transaction, 0)
	callbackFails := make([]*models.Transaction, 0)
	for _, branch := range root.Branches {
		if branch.BranchTxnStat == "2" {
			continue
		}
		if !root.IsTheFirstTimeToRecover() && branch.IsRecoverTimeExceededLimit(config.Cfg.Job.MinIntervalSeconds) {
			success = false
			callbackFails = append(callbackFails, branch)
			continue
		}
		if branch.IsRoot() && branch.IsClientSDKVersion(constant.ClientSDKVersionV1) {
			// The current version of client is V1, If the branch is root then confirm/cancel will execute by client SDK
			log.Debugsf("Skip cancel root transaction[%s]!", branch.RootXid)
			continue
		}
		branches = append(branches, branch)
		if err := callbackService.BranchTxnCallback(pair.SpanCtx, branch, false); err != nil {
			log.Errorsf("[Recover] Branch transaction cancel failed, err:%s", err)
			success = false
			branch.RecoveryTimes = branch.RecoveryTimes + 1
			branch.BranchTxnLastUpdateTime = time.Now()
			callbackFails = append(callbackFails, branch)
		}
	}
	if !success {
		if len(callbackFails) > 0 {
			firstTimeCallback := true
			if pair.Type != models.CallbackFailedRecordsRecover {
				firstTimeCallback = false
			}
			_, err := repository.CallbackFail(root.RootXid, branches, firstTimeCallback)
			if err != nil {
				log.Errorsf("[Recover] confirm callback failed, error=%++v", err)
			}
			time.Sleep(1 * time.Second)
			pair.SpanCtx.TimeoutMilliseconds = 30000
			pair.Type = models.CallbackFailedRecordsRecover
			models.Queue.Push(pair)
		}
		return
	}
	deleteTransactionByXid(repository, workerId, root.RootXid)
}

// @Desc job txn
func transactionRecoverWorker(workerId int) {
	for {
		pair := models.Queue.Pop().(models.Pair)
		repository := repo.RepositoryInstance()
		ctx := models.BuildDxcContext(context.Background(), pair.SpanCtx)
		callbackService := service.NewCallbackService(repository, client.DefaultClient, ctx)
		rootXid := pair.RootXid
		root, err := repository.FindAllBranchByRootXid(rootXid)
		if nil != err {
			log.Errorsf("Recovery workerId[%d]:Cannot found transaction with root xid:[%s]", workerId, rootXid)
			continue
		}
		switch root.RootTxnStat {
		case "0":
			recoverCancellingTransactions(repository, callbackService, workerId, root, pair)
		case "1":
			recoverConfirmingTransactions(repository, callbackService, workerId, root, pair)
		case "2":
			deleteTransactionByXid(repository, workerId, rootXid)
		case "3":
			recoverCancellingTransactions(repository, callbackService, workerId, root, pair)
		case "4":
			deleteTransactionByXid(repository, workerId, rootXid)
		}
	}
}
